package com.ds.lens.flink.monitor.monitorjob.function.detector;

import com.ds.lens.data.common.constant.CompareOperatorEnum;
import com.ds.lens.data.common.constant.MonitorStateEnum;
import com.ds.lens.data.common.document.DataPoint;
import com.ds.lens.data.common.document.dto.MonitorEvent;
import com.ds.lens.data.common.document.dynamic.DynamicMetricBO;
import com.ds.lens.data.common.document.dynamic.DynamicSinkBO;
import com.ds.lens.flink.monitor.common.BuilderUtil;
import com.ds.lens.flink.monitor.common.bo.Metric;
import com.ds.lens.flink.monitor.common.bo.MonitorDefinition;
import lombok.Builder;
import org.springframework.util.CollectionUtils;

import javax.persistence.Transient;


public class DynamicDetector implements MonitorEventDetector {
    private MonitorStateEnum state;
    private Integer atLeastTimes;
    @Transient
    private MonitorDefinition definition;
    private Double threshold;
    private CompareOperatorEnum operator;

    @Builder
    private DynamicDetector(CompareOperatorEnum operator, MonitorStateEnum state, Double threshold, Integer atLeastTimes) {
        this.state = state;
        this.atLeastTimes = atLeastTimes;
        this.threshold = threshold;
        this.operator = operator;
    }

    @Override
    public MonitorStateEnum getMonitorState() {
        return state;
    }

    @Override
    public MonitorEvent detect(Metric metric) {
        if (CollectionUtils.isEmpty(metric.getValues())) {
            return null;
        }
        int detectCount = 0;
        DynamicSinkBO dynamicSinkBO = new DynamicSinkBO();
        dynamicSinkBO.setMonitorId(metric.getMonitorKey().getMonitorId());
        dynamicSinkBO.setVersion(metric.getMonitorKey().getVersion());
        dynamicSinkBO.setMetricName(metric.getMetricName());
        for (DataPoint point : metric.getValues()) {
            if (point.getState() != null && point.getState().getLevel() > state.getLevel()) {
                detectCount++;
                continue;
            }
            Double alertThresholdUpper = point.getAlertThresholdUpper();
            Double alertThresholdLower = point.getAlertThresholdLower();
            DynamicMetricBO dynamicMetricBO = new DynamicMetricBO();
            dynamicMetricBO.setActual(point.getValue());
            dynamicMetricBO.setForecast(point.getForecast());
            dynamicMetricBO.setTimestamp(point.getTimeStamp());
            dynamicMetricBO.setAlertThresholdUpper(alertThresholdUpper);
            dynamicMetricBO.setAlertThresholdLower(alertThresholdLower);
            if (alertThresholdUpper != null && alertThresholdLower != null
                    && (point.getValue() < alertThresholdLower || point.getValue() > alertThresholdUpper)) {
                point.setState(state);
                detectCount++;
                dynamicMetricBO.setIsAnomaly(1);
            } else {
                point.setState(MonitorStateEnum.OK);
                dynamicMetricBO.setIsAnomaly(0);
            }
            dynamicSinkBO.getMetricBOS().add(dynamicMetricBO);
        }
        if (detectCount >= atLeastTimes) {
            return BuilderUtil.buildDynamicMonitorEvent(definition, metric, threshold, operator, state, dynamicSinkBO);
        } else {
            return BuilderUtil.buildDynamicMonitorEvent(definition, metric, null, null, MonitorStateEnum.OK, dynamicSinkBO);
        }
    }

    @Override
    public void setDefinition(MonitorDefinition definition) {
        this.definition = definition;
    }


    public static void main(String[] args) {
        Double actual = 2.75;
        Double forecast = 2.2418032;
        Double threshold = 3D;
        Double variance = 0.125;
//        Math.abs(actual - forecast) >= threshold * variance
        System.out.println(Math.abs(actual - forecast) >= threshold * variance);
        System.out.println(Math.abs(actual - forecast));
        System.out.println(threshold * variance);
    }

}
